package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

object Demo19Cache {
  def main(args: Array[String]): Unit = {
    /**
     * Spark比MR快的原因：
     * 1、基于内存
     * 2、DAG
     * 3、粗粒度资源调度
     */

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo19Cache")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("Spark/data/students.txt")

    val stuMapRDD: RDD[String] = stuRDD.map(line => {
      println("执行了依次map")
      line
    })

    /**
     * 对多次使用的RDD可以进行Cache缓存
     * 如果RDD的数据量非常大，内存放不下，怎么办？
     * Spark中提供了多种缓存策略，cache相当于是MEMORY_ONLY
     * 当内存放不下时可以选择：MEMORY_AND_DISK_SER策略 尽可能将缓存放入内存，序列化可以减少数据量，让更多的数据能够放入内存
     * replication副本一般只需要一个副本，如果缓存的数据丢失可以从HDFS重新读取数据并计算
     * DISK_ONLY这种情况不考虑，如果全部缓存到磁盘，不如直接从HDFS读取数据
     */
    //    stuMapRDD.cache()
    // 指定其他的缓存策略
    stuRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    // 统计班级人数
    stuMapRDD.map(line => (line.split(",")(4), 1))
      .reduceByKey(_ + _)
      .foreach(println)


    // 统计性别人数
    stuMapRDD.map(line => (line.split(",")(3), 1))
      .reduceByKey(_ + _)
      .foreach(println)

    while (true) {

    }

    // 当程序执行完后 记得释放缓存
    stuRDD.unpersist()
  }

}
